package com.atguigu.flink.sql.join;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/4/14
 */
public class Demo2_IntervalJoin
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        env.setParallelism(1);


        SingleOutputStreamOperator<WaterSensor> ds1 = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());

        SingleOutputStreamOperator<WaterSensor> ds2 = env
            .socketTextStream("hadoop102", 8889)
            .map(new WaterSensorMapFunction());

        Schema schema = Schema.newBuilder()
                              .column("id", "STRING")
                              .column("ts", "BIGINT")
                              .column("vc", "INT")
                              .columnByExpression("pt", "proctime()")
                              .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts,3)")
                              .watermark("et","et - INTERVAL '0.001' SECOND")
                              .build();
        //从流中获取时间属性
        tableEnvironment.createTemporaryView("t1",ds1,schema);
        tableEnvironment.createTemporaryView("t2",ds2,schema);

        /*
                借助水印
         */
        tableEnvironment.sqlQuery(" select t1.id,t1.et,t1.vc ,t2.id,t2.et,t2.vc " +
            "  from t1,t2 " +
            // 指定关联条件
            "  where  t1.id = t2.id and " +
            //取t2的数据的时间在   t1时间的 前后2s 内到达
            "   t2.et between t1.et - INTERVAL '2' SECOND" +
            "         and   t1.et + INTERVAL '2' SECOND ")
                        .execute()
                        .print();

    }
}
